// ============================================
// 18 - 并发编程 (Concurrency)
// ============================================

/// Rust 提供了安全的并发编程支持

use std::thread;
use std::time::Duration;
use std::sync::{mpsc, Mutex, Arc};

// ==================== 线程基础 ====================

/// 创建线程
pub fn creating_threads() {
    println!("\n=== 创建线程 ===");
    
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("  子线程: {}", i);
            thread::sleep(Duration::from_millis(1));
        }
    });
    
    for i in 1..=3 {
        println!("主线程: {}", i);
        thread::sleep(Duration::from_millis(1));
    }
    
    // 等待线程完成
    handle.join().unwrap();
    println!("线程已完成");
}

/// 使用 move 闭包
pub fn move_closures() {
    println!("\n=== move 闭包 ===");
    
    let v = vec![1, 2, 3];
    
    let handle = thread::spawn(move || {
        println!("向量: {:?}", v);
    });
    
    handle.join().unwrap();
    // println!("{:?}", v); // ❌ 错误: v 已被移动
}

// ==================== 消息传递 ====================

/// 使用消息传递在线程间传输数据
pub fn message_passing() {
    println!("\n=== 消息传递 ===");
    
    let (tx, rx) = mpsc::channel();
    
    thread::spawn(move || {
        let val = String::from("你好");
        tx.send(val).unwrap();
    });
    
    let received = rx.recv().unwrap();
    println!("收到: {}", received);
}

/// 发送多个值
pub fn sending_multiple_values() {
    println!("\n=== 发送多个值 ===");
    
    let (tx, rx) = mpsc::channel();
    
    thread::spawn(move || {
        let vals = vec![
            String::from("你好"),
            String::from("来自"),
            String::from("线程"),
        ];
        
        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    for received in rx {
        println!("收到: {}", received);
    }
}

/// 多个生产者
pub fn multiple_producers() {
    println!("\n=== 多个生产者 ===");
    
    let (tx, rx) = mpsc::channel();
    let tx1 = tx.clone();
    
    thread::spawn(move || {
        let vals = vec![
            String::from("线程1: 消息1"),
            String::from("线程1: 消息2"),
        ];
        
        for val in vals {
            tx1.send(val).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    thread::spawn(move || {
        let vals = vec![
            String::from("线程2: 消息1"),
            String::from("线程2: 消息2"),
        ];
        
        for val in vals {
            tx.send(val).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    for received in rx {
        println!("收到: {}", received);
    }
}

// ==================== 共享状态 ====================

/// Mutex<T> - 互斥锁
pub fn mutex_basics() {
    println!("\n=== Mutex<T> ===");
    
    let m = Mutex::new(5);
    
    {
        let mut num = m.lock().unwrap();
        *num = 6;
    } // 锁在这里自动释放
    
    println!("m = {:?}", m);
}

/// 多线程共享 Mutex<T>
pub fn sharing_mutex() {
    println!("\n=== 共享 Mutex ===");
    
    let counter = Arc::new(Mutex::new(0));
    let mut handles = vec![];
    
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            let mut num = counter.lock().unwrap();
            *num += 1;
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("结果: {}", *counter.lock().unwrap());
}

// ==================== RwLock<T> ====================

/// RwLock<T> - 读写锁
pub fn rwlock_basics() {
    println!("\n=== RwLock<T> ===");
    
    use std::sync::RwLock;
    
    let lock = RwLock::new(5);
    
    // 多个读者
    {
        let r1 = lock.read().unwrap();
        let r2 = lock.read().unwrap();
        println!("r1 = {}, r2 = {}", r1, r2);
    }
    
    // 单个写者
    {
        let mut w = lock.write().unwrap();
        *w += 1;
    }
    
    println!("lock = {:?}", lock);
}

// ==================== 线程池 ====================

/// 线程池概念
pub fn thread_pool_concept() {
    println!("\n=== 线程池 ===");
    
    println!("线程池维护一组线程，重复使用");
    println!("避免频繁创建和销毁线程的开销");
    println!("可以使用 rayon 等 crate 实现线程池");
}

// ==================== Send 和 Sync Trait ====================

/// Send 和 Sync trait
pub fn send_and_sync() {
    println!("\n=== Send 和 Sync ===");
    
    println!("Send trait:");
    println!("  - 允许在线程间转移所有权");
    println!("  - 大多数类型实现了 Send");
    println!("  - Rc<T> 没有实现 Send");
    
    println!("\nSync trait:");
    println!("  - 允许在线程间共享引用");
    println!("  - T 是 Sync 则 &T 可以安全地在线程间传递");
    println!("  - RefCell<T> 和 Cell<T> 没有实现 Sync");
}

// ==================== 原子类型 ====================

/// 原子类型
pub fn atomic_types() {
    println!("\n=== 原子类型 ===");
    
    use std::sync::atomic::{AtomicUsize, Ordering};
    
    let counter = Arc::new(AtomicUsize::new(0));
    let mut handles = vec![];
    
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            for _ in 0..1000 {
                counter.fetch_add(1, Ordering::SeqCst);
            }
        });
        handles.push(handle);
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
    
    println!("原子计数器结果: {}", counter.load(Ordering::SeqCst));
}

// ==================== 死锁 ====================

/// 死锁问题
pub fn deadlock_problem() {
    println!("\n=== 死锁 ===");
    
    println!("死锁发生条件:");
    println!("1. 互斥条件");
    println!("2. 持有并等待");
    println!("3. 不可抢占");
    println!("4. 循环等待");
    
    println!("\n避免死锁:");
    println!("1. 统一锁顺序");
    println!("2. 使用超时");
    println!("3. 使用 try_lock");
}

// ==================== Barrier ====================

/// 屏障同步
pub fn barrier_sync() {
    println!("\n=== Barrier ===");
    
    use std::sync::Barrier;
    
    let mut handles = Vec::with_capacity(3);
    let barrier = Arc::new(Barrier::new(3));
    
    for i in 0..3 {
        let c = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("线程 {} 开始工作", i);
            thread::sleep(Duration::from_millis(100 * i as u64));
            println!("线程 {} 到达屏障", i);
            c.wait();
            println!("线程 {} 继续执行", i);
        }));
    }
    
    for handle in handles {
        handle.join().unwrap();
    }
}

// ==================== 条件变量 ====================

/// 条件变量
pub fn condvar_basics() {
    println!("\n=== 条件变量 ===");
    
    use std::sync::Condvar;
    
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair2 = Arc::clone(&pair);
    
    thread::spawn(move || {
        thread::sleep(Duration::from_secs(1));
        let (lock, cvar) = &*pair2;
        let mut started = lock.lock().unwrap();
        *started = true;
        cvar.notify_one();
        println!("子线程通知");
    });
    
    let (lock, cvar) = &*pair;
    let mut started = lock.lock().unwrap();
    while !*started {
        println!("主线程等待...");
        started = cvar.wait(started).unwrap();
    }
    println!("主线程继续");
}

// ==================== 线程局部存储 ====================

/// 线程局部存储
pub fn thread_local_storage() {
    println!("\n=== 线程局部存储 ===");
    
    use std::cell::RefCell;
    
    thread_local! {
        static COUNTER: RefCell<u32> = RefCell::new(0);
    }
    
    COUNTER.with(|c| {
        *c.borrow_mut() += 1;
        println!("主线程计数器: {}", c.borrow());
    });
    
    thread::spawn(|| {
        COUNTER.with(|c| {
            *c.borrow_mut() += 1;
            println!("子线程计数器: {}", c.borrow());
        });
    }).join().unwrap();
}

// ==================== 实际应用示例 ====================

/// 并行计算
pub fn parallel_computation() {
    println!("\n=== 并行计算 ===");
    
    let data = vec![1, 2, 3, 4, 5, 6, 7, 8];
    let chunk_size = data.len() / 4;
    let (tx, rx) = mpsc::channel();
    
    for chunk in data.chunks(chunk_size) {
        let tx = tx.clone();
        let chunk = chunk.to_vec();
        thread::spawn(move || {
            let sum: i32 = chunk.iter().sum();
            tx.send(sum).unwrap();
        });
    }
    
    drop(tx); // 关闭发送端
    
    let total: i32 = rx.iter().sum();
    println!("总和: {}", total);
}

/// 生产者-消费者模式
pub fn producer_consumer() {
    println!("\n=== 生产者-消费者 ===");
    
    let (tx, rx) = mpsc::channel();
    
    // 生产者
    thread::spawn(move || {
        for i in 0..5 {
            println!("生产: {}", i);
            tx.send(i).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });
    
    // 消费者
    thread::spawn(move || {
        for received in rx {
            println!("  消费: {}", received);
            thread::sleep(Duration::from_millis(150));
        }
    }).join().unwrap();
}

// ==================== 并发最佳实践 ====================

/// 并发最佳实践
pub fn concurrency_best_practices() {
    println!("\n=== 并发最佳实践 ===");
    
    println!("1. 优先使用消息传递而非共享状态");
    println!("2. 使用 Arc<Mutex<T>> 共享可变状态");
    println!("3. 避免死锁 - 统一锁顺序");
    println!("4. 使用原子类型处理简单计数");
    println!("5. 考虑使用线程池");
    println!("6. 注意 Send 和 Sync 的含义");
    println!("7. 使用 Rust 的类型系统防止数据竞争");
    println!("8. 测试并发代码的各种情况");
}

/// 主函数 - 演示所有功能
#[allow(dead_code)]
pub fn main() {
    println!("╔════════════════════════════════════════╗");
    println!("║   Rust 并发编程教程                    ║");
    println!("╚════════════════════════════════════════╝");
    
    creating_threads();
    move_closures();
    message_passing();
    sending_multiple_values();
    multiple_producers();
    mutex_basics();
    sharing_mutex();
    rwlock_basics();
    thread_pool_concept();
    send_and_sync();
    atomic_types();
    deadlock_problem();
    barrier_sync();
    condvar_basics();
    thread_local_storage();
    parallel_computation();
    producer_consumer();
    concurrency_best_practices();
    
    println!("\n✅ 并发编程教程完成！");
}

#[cfg(test)]
mod tests {
    use super::*;
    
    #[test]
    fn test_message_passing() {
        let (tx, rx) = mpsc::channel();
        tx.send(10).unwrap();
        assert_eq!(rx.recv().unwrap(), 10);
    }
    
    #[test]
    fn test_mutex() {
        let m = Mutex::new(5);
        {
            let mut num = m.lock().unwrap();
            *num = 6;
        }
        assert_eq!(*m.lock().unwrap(), 6);
    }
    
    #[test]
    fn test_arc_mutex() {
        let counter = Arc::new(Mutex::new(0));
        let counter2 = Arc::clone(&counter);
        
        thread::spawn(move || {
            let mut num = counter2.lock().unwrap();
            *num += 1;
        }).join().unwrap();
        
        assert_eq!(*counter.lock().unwrap(), 1);
    }
}

